package com.chinamobile.iot.handler;

import com.alibaba.fastjson.JSON;
import com.chinamobile.iot.config.BrokerConfiguration;
import com.chinamobile.iot.config.MqttConfiguration;
import com.chinamobile.iot.config.ZookeeperConfiguration;
import com.chinamobile.iot.handler.impl.BrokerMonitorImpl;
import com.chinamobile.iot.handler.impl.ZkHelper;
import com.chinamobile.iot.handler.model.*;
import com.chinamobile.iot.handler.redis.RedisListener;
import com.chinamobile.iot.handler.redis.RedisService;
import com.chinamobile.iot.mqtt.interception.MessageHandler;
import com.chinamobile.iot.mqtt.spi.impl.ProtocolProcessor;
import com.chinamobile.iot.mqtt.spi.impl.SubscriptionTree;
import com.chinamobile.iot.mqtt.spi.model.Qos1Message;
import com.chinamobile.iot.mqtt.spi.model.StoredMessage;
import com.chinamobile.iot.mqtt.spi.model.Tree1Client;
import com.chinamobile.iot.rest.BrokerClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.LinkedList;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import static com.chinamobile.iot.handler.model.SubscriptionMessage.TYPE_SUBSCRIBE;
import static com.chinamobile.iot.handler.model.SubscriptionMessage.TYPE_UNSUBSCIRBE;
import static com.chinamobile.iot.handler.redis.RedisConst.*;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE;

/**
 * Created by szl on 2017/7/31.
 */
@Component("extendMessageHandler")
@ConditionalOnExpression("${mqtt.distribute}")
public class DistributeMessageHandler implements MessageHandler, ZkHelper.ZkConnectEventHandler {

    private static Logger logger = LoggerFactory.getLogger(DistributeMessageHandler.class);
    @Autowired
    private ZookeeperConfiguration zooConfiguration;
    @Autowired
    private BrokerConfiguration brokerConfiguration;
    @Autowired
    private MqttConfiguration mqttConfiguration;

    @Value("${server.address}")
    private String restIp;
    @Value("${server.port}")
    private int restPort;

    /**
     * 本地使用的订阅树
     */
    private SubscriptionTree<Tree1Client> tree1;
    /**
     * the subscription tree of other broker subscribe, used for CleanSession = true
     */
    private SubscriptionTree<String> tree2;
    /**
     * the subscription tree of global, used for CleanSession = false
     */
    private SubscriptionTree<Tree3Client> tree3;
    /**
     * zookeeper helper class
     */
    private ZkHelper zkHelper;
    @Autowired
    private BrokerMonitorImpl brokerMonitor;
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private RedisService redisService;
    @Autowired
    private RedisListener redisListener;
    @Autowired
    private BrokerClient brokerClient;

    /**
     * the status of zkService, in onConnecting stat, when receive mqtt unsubscribe message, it just drop it
     */
    private volatile boolean init = false;

    private volatile boolean ok = false;

    /**
     * used by addAllClientToZk, hasClientConnected
     */
    @Autowired
    private ProtocolProcessor protocolProcessor;


    @Override
    public boolean onConnected(ZkHelper zkHelper) {
        init = true;
        try {
            if (!addSelfToZookeeper()) {
                BrokerStat stat = getSelfBrokerStat();
                switch (stat) {
                    case NORMAL: {
                        // 闪段的情况，什么都不用做
                        return true;
                    }
                    case NOEXIST: {
                        return false;
                    }
                    case OLD: {
                        // 可能broker重启了，但是老结点还在，让其触发管理员结点事件
                        logger.info("old node with the same name is find, please wait the old node to stoped");
                        return false;
                    }
                    case MAINTAIN: {
                        // 当前管理员正在维护，直接退出
                        logger.info("maintain node is find, waiting...");
                        return false;
                    }
                }
            } else {
                logger.info("restart redis subscription");
                addAllClientToZookeeper();
                if (mqttConfiguration.isSubscribe()) {
                    // 尝试成为broker信息维护者
                    logger.info("try to become manager");
                    tryBecomeManager();
                    brokerMonitor.setManagerBroker(zkHelper.getManager());
                    stopRedisSubscribe();
                    startRedisSubscribe();
                    // 读取所有存量的订阅信息
                    logger.info("read the whole subscription in the redis");
                    readRedisSubscriptions();

                }
            }
        } catch (Exception e) {
            logger.error("onConnected failed. {}", e.getMessage());
        }
        return true;
    }

    @Override
    public boolean init() {
        logger.info("DistributeMessageHandler::init() is called");
        try {
            tree1 = protocolProcessor.getTree1();
            if (mqttConfiguration.isSubscribe()) {
                tree2 = new SubscriptionTree<>();
                tree3 = new SubscriptionTree<>();
                // brokerMonitor use tree2 to remove subscription when the broker is offline
                brokerMonitor.setTree2(tree2);
                brokerMonitor.setRedisTemplate(redisTemplate);
                // set the subscription tree that the redis listener will modify it
                redisListener.setTree2(tree2);
                redisListener.setTree3(tree3);
                // begin listen the redis channel
                redisService.addListener(CHANNEL_SUBSCRIPTION);
            }

            // init the zkHelper and start watch the znodes, if connect to zookeeper is ok, the callback onConnect will be called
            zkHelper = new ZkHelper(zooConfiguration.getServers(), brokerMonitor, this, brokerConfiguration.getName());
            zkHelper.init();
        } catch (Exception e) {
            ok = false;
            return false;
        }
        ok = true;
        return true;
    }

    @Override
    public void destroy() {

    }

    @Override
    public boolean onConnecting(String clientId, String mqttId) {
        // 如果添加新的Client失败，让对应的客户端下线
        if (!addClient(clientId, mqttId)) {
            ZkClientInfo client = getClient(mqttId);
            if (client != null) {
                if (restIp.equals(client.getIp()) && restPort == client.getPort()){
                    protocolProcessor.offlineMqttId(mqttId);
                }
                else{
                    brokerClient.offlineDevice(client.getIp(), client.getPort(), mqttId);
                }

                //removeClient(mqttId);
            }
            // 再次下线，如果还不能成功，则表明这段时间内有其他客户端上线，本客户端应该退出
            if (!addClient(clientId, mqttId)) {
                return false;
            }
        }
        return true;
    }

    @Override
    public void onConnected(String clientId, String mqttId, boolean cleanSession) {
        // 读取以前是否有保存的QoS订阅
        if (!mqttConfiguration.isSubscribe() || !mqttConfiguration.isCleanSession()) {
            return;
        }
        try {
            SetOperations<String, String> setOperations = redisTemplate.opsForSet();
            String redis3Key = REDIS_3_KEY_PREFIX + clientId;
            Set<String> result = setOperations.members(redis3Key);

            if (cleanSession) {
                if (result != null && !result.isEmpty()) {
                    result.forEach((sub) -> {
                        RedisClientSessionSub redisClientSessionSub = JSON.parseObject(sub, RedisClientSessionSub.class);
                        int qos = redisClientSessionSub.getQ();
                        String filter = redisClientSessionSub.getF();
                        String broker = redisClientSessionSub.getB();

                        // 加入本地订阅树
                        Tree1Client client1 = new Tree1Client(clientId, true);
                        tree1.add(filter, client1, qos);
                        Tree3Client client3 = new Tree3Client(clientId, broker);
                        tree3.removeSubscribe(client3);

                        // 给其他broker广播发送去订阅消息
                        SubscriptionMessage message = new SubscriptionMessage();
                        message.setClientId(clientId);
                        message.setBroker(broker);
                        message.setQos(qos);
                        message.setFilter(filter);
                        message.setType(TYPE_UNSUBSCIRBE);
                        message.setCleanSession(false);
                        redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));

                        // 在Redis2中删除全局订阅
                        RedisBrokerSessionSub redisBrokerSessionSub = new RedisBrokerSessionSub();
                        redisBrokerSessionSub.setB(broker);
                        redisBrokerSessionSub.setQ(qos);
                        redisBrokerSessionSub.setF(filter);
                        redisBrokerSessionSub.setC(clientId);
                        setOperations.remove(REDIS_2_KEY_GLOBAL, JSON.toJSONString(redisBrokerSessionSub));
                        // 在Redis1中添加记录
                        setOperations.add(brokerConfiguration.getName(), filter);

                        // 给其他broker发送订阅消息
                        message.setBroker(brokerConfiguration.getName());
                        message.setType(SubscriptionMessage.TYPE_SUBSCRIBE);
                        message.setCleanSession(true);
                        redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
                    });
                    redisTemplate.delete(redis3Key);
                }

            } else {
                LinkedList<String> newSub = new LinkedList<>();
                if (result != null && !result.isEmpty()) {
                    result.forEach((sub) -> {
                        RedisClientSessionSub redisClientSessionSub = JSON.parseObject(sub, RedisClientSessionSub.class);

                        int qos = redisClientSessionSub.getQ();
                        String filter = redisClientSessionSub.getF();
                        String broker = redisClientSessionSub.getB();

                        Tree1Client client1 = new Tree1Client(clientId, false);
                        tree1.add(filter, client1, qos);
                        Tree3Client client3 = new Tree3Client(clientId, broker);
                        tree3.removeSubscribe(client3);

                        // 发送去掉全局订阅的消息
                        SubscriptionMessage message = new SubscriptionMessage();
                        message.setClientId(clientId);
                        message.setBroker(broker);
                        message.setQos(qos);
                        message.setFilter(filter);
                        message.setType(TYPE_UNSUBSCIRBE);
                        message.setCleanSession(false);


                        //在 Redis2中先删除原来的记录，再新增记录达到更新的效果
                        RedisBrokerSessionSub redisBrokerSessionSub = new RedisBrokerSessionSub();
                        redisBrokerSessionSub.setB(broker);
                        redisBrokerSessionSub.setQ(qos);
                        redisBrokerSessionSub.setF(filter);
                        redisBrokerSessionSub.setC(clientId);
                        setOperations.remove(REDIS_2_KEY_GLOBAL, JSON.toJSONString(redisBrokerSessionSub));
                        redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));

                        redisBrokerSessionSub.setB(brokerConfiguration.getName());
                        setOperations.add(REDIS_2_KEY_GLOBAL, JSON.toJSONString(redisBrokerSessionSub));

                        // 新增订阅
                        message.setBroker(brokerConfiguration.getName());
                        message.setType(SubscriptionMessage.TYPE_SUBSCRIBE);
                        message.setCleanSession(false);
                        redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));

                        redisClientSessionSub.setB(brokerConfiguration.getName());
                        newSub.add(JSON.toJSONString(redisClientSessionSub));
                    });
                    // 在redis3先删除记录，再新增，达到修改其broker信息的效果
                    redisTemplate.delete(redis3Key);
                    newSub.forEach((s) -> setOperations.add(redis3Key, s));
                }
            }
        } catch (Exception e) {
            logger.info("error occured when onConnected is called. clientId = {}, message = {}", clientId, e.getMessage());
        }
    }

    @Override
    public void onClientDisconnected(String clientId, String mqttId, boolean cleanSession) {
        if (clientId == null) {
            return;
        }
        disconnectClient(clientId, mqttId, cleanSession);
    }


    @Override
    public void onTopicPublished(String clientId, String mqttId, StoredMessage storedMessage, String filter, int subCount) {

        if (!mqttConfiguration.isSubscribe()) {
            return;
        }

        try {
            String topic = storedMessage.getTopic();
            // 如果不是其他broker发送过来的消息，检查tree2，看是否远端其他broker需要本消息
            if (clientId != null && mqttId != null) {
                Set<SubscriptionTree.ClientInfo<String>> clientInfos = tree2.matchFull(topic);
                if (clientInfos != null && !clientInfos.isEmpty()) {
                    clientInfos.forEach((c) -> {
                        BrokerInfo brokerInfo = brokerMonitor.getMapBrokerInfo().get(c.getClient());
                        if (brokerInfo != null) {
                            brokerClient.sendSubMessage(brokerInfo.getIp(), brokerInfo.getPort(), c.getFilter(), storedMessage);
                        }
                    });
                }
            }

            // 检查tree3
            Set<SubscriptionTree.ClientInfo<Tree3Client>> tree3Clients = tree3.matchFull(topic);
            if (tree3Clients != null && !tree3Clients.isEmpty()) {
                tree3Clients.forEach((c) -> {
                    Tree3Client tree3Client = c.getClient();
                    int qos = storedMessage.getQos().value() < c.getQos() ? storedMessage.getQos().value() : c.getQos();
                    if (clientId != null && mqttId != null) {
                        if (tree3Client.getB() != null) {
                            BrokerInfo brokerInfo = brokerMonitor.getMapBrokerInfo().get(tree3Client.getB());
                            if (brokerInfo != null && !brokerInfo.getName().equals(brokerConfiguration.getName())) {
                                boolean ok = brokerClient.sendSubMessage(brokerInfo.getIp(), brokerInfo.getPort(), c.getFilter(), storedMessage);
                                if (!ok && qos == AT_LEAST_ONCE.value()) {
                                    saveMsgForClient(tree3Client.getC(), storedMessage, storedMessage.getMessageId(), 1);
                                }
                            } else if (qos == AT_LEAST_ONCE.value()) {
                                // 包数据保存在redis中
                                saveMsgForClient(tree3Client.getC(), storedMessage, storedMessage.getMessageId(), 1);
                            }
                        } else {
                            if (qos == AT_LEAST_ONCE.value()) {
                                saveMsgForClient(tree3Client.getC(), storedMessage, storedMessage.getMessageId(), 1);
                            }
                        }
                    } else if (tree3Client.getB() == null) {
                        // 如果clientId或者mqttId为null，说明是其他Broker发送过来的消息，可能是本broker某个cleansession为false的
                        // client刚刚断线，其他broker还未收到redis去订阅消息或者其他broker与redis之间的访问出现异常，此时应重新广播删除
                        // 旧订阅，增加新订阅的
                        SubscriptionMessage message = new SubscriptionMessage();
                        message.setClientId(tree3Client.getC());
                        message.setBroker(brokerConfiguration.getName());
                        message.setFilter(c.getFilter());
                        message.setCleanSession(false);
                        message.setQos(c.getQos());
                        message.setType(TYPE_UNSUBSCIRBE);
                        redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
                        message.setBroker(null);
                        message.setType(TYPE_SUBSCRIBE);
                        redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
                    }
                });
            } else {
                // 如果本次发送过来的消息没有任何一个本地客户端或者保存回话的客户端需要，应广播取消此filter的订阅，让其他filter不要再次发送
                if (subCount == 0 && clientId == null) {
                    SubscriptionMessage message = new SubscriptionMessage();
                    message.setClientId(null);
                    message.setBroker(brokerConfiguration.getName());
                    message.setFilter(filter);
                    message.setCleanSession(true);
                    message.setQos(storedMessage.getQos().value());
                    message.setType(TYPE_UNSUBSCIRBE);
                    redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
                }
            }
        } catch (Exception e) {
            logger.error("error occur when onTopicPublished(), clientId = {}, topic = {}, message = {}", clientId, filter, e.getMessage());
        }
    }

    @Override
    public void onTopicSubscribed(String clientId, String mqttId, String filter, int qos, boolean cleanSession, int oldQos) {
        if (!mqttConfiguration.isSubscribe()) {
            return;
        }
        SetOperations<String, String> setOperations = redisTemplate.opsForSet();
        if (cleanSession) {
            // 在redis1中添加记录
            setOperations.add(brokerConfiguration.getName(), filter);
        } else {
            // 在Redis3中添加记录
            String clientKey = REDIS_3_KEY_PREFIX + clientId;
            RedisClientSessionSub clientSub = new RedisClientSessionSub();
            clientSub.setB(brokerConfiguration.getName());
            clientSub.setQ(qos);
            clientSub.setF(filter);
            setOperations.add(clientKey, JSON.toJSONString(clientSub));


            // 在Redis2中添加记录
            RedisBrokerSessionSub brokerSub = new RedisBrokerSessionSub();
            brokerSub.setB(brokerConfiguration.getName());
            brokerSub.setC(clientId);
            brokerSub.setF(filter);
            brokerSub.setQ(qos);
            setOperations.add(REDIS_2_KEY_GLOBAL, JSON.toJSONString(brokerSub));
            // 如果存在旧的qos，就删掉旧的qos
            if (oldQos != 3 && oldQos != qos){
                clientSub.setQ(oldQos);
                setOperations.remove(clientKey, JSON.toJSONString(clientSub));
                brokerSub.setQ(oldQos);
                setOperations.remove(REDIS_2_KEY_GLOBAL, JSON.toJSONString(brokerSub));
            }
        }
        // 给其他broker发送广播消息
        SubscriptionMessage message = new SubscriptionMessage();
        message.setBroker(brokerConfiguration.getName());
        message.setCleanSession(cleanSession);
        message.setClientId(clientId);
        message.setFilter(filter);
        if (oldQos != 3 && oldQos != qos){
            message.setQos(oldQos);
            message.setType(TYPE_UNSUBSCIRBE);
            redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
        }
        message.setQos(qos);
        message.setType(SubscriptionMessage.TYPE_SUBSCRIBE);
        redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
    }

    @Override
    public void onTopicUnsubscribed(String filter, int qos, String clientId, boolean cleanSession) {
        if (!mqttConfiguration.isSubscribe()) {
            return;
        }
        // 删掉Redis1
        SetOperations<String, String> setOperations = redisTemplate.opsForSet();
        if (cleanSession) {
            // 如果没有其他client subscribe这个filter，就从redis1中删掉
            if (canDeleteFromRedis1(filter)){
                // 从redis1中删除订阅信息
                setOperations.remove(brokerConfiguration.getName(), filter);

                // 发送广播
                // 给其他broker发送广播消息
                SubscriptionMessage message = new SubscriptionMessage();
                message.setBroker(brokerConfiguration.getName());
                message.setCleanSession(true);
                message.setClientId(clientId);
                message.setFilter(filter);
                message.setQos(qos);
                message.setType(TYPE_UNSUBSCIRBE);
                redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
            }
        } else {
            // 从Redis2中删除订阅信息
            RedisBrokerSessionSub brokerSessionSub = new RedisBrokerSessionSub();
            brokerSessionSub.setF(filter);
            brokerSessionSub.setQ(qos);
            brokerSessionSub.setC(clientId);
            brokerSessionSub.setB(brokerConfiguration.getName());
            setOperations.remove(REDIS_2_KEY_GLOBAL, JSON.toJSONString(brokerSessionSub));

            // 从Redis3中删除订阅信息
            RedisClientSessionSub clientSessionSub = new RedisClientSessionSub();
            clientSessionSub.setB(brokerConfiguration.getName());
            clientSessionSub.setF(filter);
            clientSessionSub.setQ(qos);
            String key = REDIS_3_KEY_PREFIX + clientId;
            setOperations.remove(key, JSON.toJSONString(clientSessionSub));

            // 给其他broker发送广播消息
            SubscriptionMessage message = new SubscriptionMessage();
            message.setBroker(brokerConfiguration.getName());
            message.setCleanSession(false);
            message.setClientId(clientId);
            message.setFilter(filter);
            message.setQos(qos);
            message.setType(TYPE_UNSUBSCIRBE);
            redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
        }
    }

    @Override
    public void onMessageAcknowledged(String clientId, String mqttId, int messageId) {

    }

    @Override
    public void onRouteMessageToClient(String clientId, String mqttId, String topic) {

    }

    /**
     * add self broker to zookeeper
     *
     * @return true if success, false if failed.
     */
    private boolean addSelfToZookeeper() {
        if (!init) {
            return false;
        }
        String name = brokerConfiguration.getName();
        BrokerInfo brokerInfo = new BrokerInfo();
        brokerInfo.setName(name);
        brokerInfo.setIp(brokerConfiguration.getHost());
        brokerInfo.setPort(brokerConfiguration.getPort());
        brokerInfo.setStartTime(new Date());
        return zkHelper.addBroker(name, brokerInfo);
    }

    private BrokerStat getSelfBrokerStat() {
        if (!init) {
            return null;
        }
        return zkHelper.getBrokerStat(brokerConfiguration.getName());
    }

    private void removeSelfBroker() {
        if (!init) {
            return;
        }
        zkHelper.removeBroker(brokerConfiguration.getName());
    }

    private void addAllClientToZookeeper() {
        protocolProcessor.getMqttIdMap().forEach((k, v) -> addClient(v, k));
    }

    private boolean addClientToZk(String clientId, String mqttId) {
        // 如果没有启用zkService时，默认都返回true
        if (!init) {
            return true;
        }

        ZkClientInfo clientInfo = new ZkClientInfo();
        clientInfo.setC(clientId);
        clientInfo.setIp(restIp);
        clientInfo.setPort(restPort);
        clientInfo.setB(brokerConfiguration.getName());
        return zkHelper.addClient(mqttId, clientInfo);
    }

    private void startRedisSubscribe() {
        if (!init) {
            return;
        }
        redisService.addListener(CHANNEL_SUBSCRIPTION);
    }

    private void stopRedisSubscribe() {
        if (!init) {
            return;
        }
        redisService.removeListen(CHANNEL_SUBSCRIPTION);
    }

    private void readRedisSubscriptions() {
        if (!init) {
            return;
        }
        ConcurrentHashMap<String, BrokerInfo> map = brokerMonitor.getMapBrokerInfo();
        map.forEach((k, v) -> {
            SetOperations<String, String> setOperations = redisTemplate.opsForSet();
            Set<String> filters = setOperations.members(k);
            if (filters != null && !filters.isEmpty()){
                filters.forEach((filter) -> {
                    tree2.add(filter, k, 1);
                });
            }
        });

        // 如果当前启用了cleanSession功能，则读取全局订阅记录，否则不用读取
        if (mqttConfiguration.isCleanSession()) {
            SetOperations<String, String> setOperations = redisTemplate.opsForSet();
            Set<String> globalFilters = setOperations.members(REDIS_2_KEY_GLOBAL);
            globalFilters.forEach((f) -> {
                RedisBrokerSessionSub sub = JSON.parseObject(f, RedisBrokerSessionSub.class);
                Tree3Client tree3Client = new Tree3Client(sub.getC(), sub.getB());
                tree3.add(sub.getF(), tree3Client, sub.getQ());
            });
        }


        redisListener.setNormalMode();
    }


    private boolean isConnectedToZookeeper() {
        return init && zkHelper.isConnectedToZookeeper();
    }

    private boolean tryBecomeManager() {
        if (!init) {
            return false;
        }
        return zkHelper.tryBecomeManager(brokerConfiguration.getName());
    }



    private boolean addClient(String clientId, String mqttId) {
        logger.info("start addClient clientId = {}, mqttId = {}", clientId, mqttId);
        return addClientToZk(clientId, mqttId);
    }

    private ZkClientInfo getClient(String mqttId) {
        if (!init) {
            return null;
        }
        return zkHelper.getClient(mqttId);
    }

    public void removeClient(String mqttId) {
        logger.info("start removeClient mqttId = {}", mqttId);
        if (!init) {
            return;
        }
        zkHelper.removeClient(mqttId);
    }

    private void saveMsgForClient(String clientId, StoredMessage message, Integer messageId, int round) {
        Qos1Message qos1Message = new Qos1Message();
        qos1Message.setRound(round);
        qos1Message.setToClientId(clientId);
        qos1Message.setMsgId(messageId);
        qos1Message.setData(message);

        String key = REDIS_4_KEY_PREFIX + clientId;

        ListOperations<String, String> listOperations = redisTemplate.opsForList();
        long size = listOperations.size(key);
        if (size == 0) {
            listOperations.rightPush(key, JSON.toJSONString(qos1Message));
            redisTemplate.expire(key, 3, TimeUnit.DAYS);
        } else if (size < 500) {
            listOperations.rightPush(key, JSON.toJSONString(qos1Message));
        }
    }



    private void disconnectClient(String clientId, String mqttId, boolean cleanSession) {
        if (!init) {
            return;
        }
        if (mqttConfiguration.isSubscribe()) {
            try {
                SetOperations<String, String> setOperations = redisTemplate.opsForSet();
                if (!cleanSession) {
                    // 读取redis3中该客户端的所有全局订阅
                    String redis3Key = REDIS_3_KEY_PREFIX + clientId;
                    Set<String> subs = setOperations.members(redis3Key);
                    LinkedList<String> clientSessionSubs = new LinkedList<>();
                    if (subs != null && subs.size() > 0) {
                        subs.forEach(sub -> clientSessionSubs.add(sub));
                    }

                    clientSessionSubs.forEach((s) -> {
                        RedisClientSessionSub sub = JSON.parseObject(s, RedisClientSessionSub.class);

                        // 修改tree3中的记录，设置其
                        Tree3Client client3 = new Tree3Client(clientId, sub.getB());
                        tree3.removeSubscribe(client3);
                        client3.setB(null);
                        tree3.add(sub.getF(), client3, sub.getQ());

                        // 在tree1中删掉订阅记录
                        Tree1Client client1 = new Tree1Client(clientId, false);
                        tree1.removeSubscribe(client1);

                        RedisBrokerSessionSub brokerSessionSub = new RedisBrokerSessionSub();
                        brokerSessionSub.setB(sub.getB());
                        brokerSessionSub.setC(clientId);
                        brokerSessionSub.setQ(sub.getQ());
                        brokerSessionSub.setF(sub.getF());

                        SubscriptionMessage message = new SubscriptionMessage();
                        message.setBroker(sub.getB());
                        message.setFilter(sub.getF());
                        message.setQos(sub.getQ());
                        message.setClientId(clientId);
                        message.setCleanSession(false);
                        message.setType(TYPE_UNSUBSCIRBE);

                        // 删掉Redis2中的记录
                        setOperations.remove(REDIS_2_KEY_GLOBAL, JSON.toJSONString(brokerSessionSub));
                        //在redis2中添加新记录
                        brokerSessionSub.setB(null);
                        setOperations.add(REDIS_2_KEY_GLOBAL, JSON.toJSONString(brokerSessionSub));

                        // 修改redis3中所有记录，设置其broker为null
                        setOperations.remove(redis3Key, JSON.toJSONString(sub));
                        sub.setB(null);
                        setOperations.add(redis3Key, JSON.toJSONString(sub));

                        // 向其他broker广播取消订阅消息
                        redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
                        // 向其他broker广播新建订阅消息，这里设置broker为null
                        message.setBroker(null);
                        message.setType(TYPE_SUBSCRIBE);
                        redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
                    });
                } else {
                    // cleansession = true
                    // 从tree1中查询有哪些Topic本
                    if (clientId != null && !clientId.isEmpty()) {
                        Tree1Client client1 = new Tree1Client(clientId, true);
                        ConcurrentHashMap<String, Integer> map = tree1.getSubscription(client1);
                        if (map != null) {
                            tree1.removeSubscribe(client1);
                            map.forEach((f, q) -> {
                                if (canDeleteFromRedis1(f)){
                                    // 删除redis1
                                    setOperations.remove(brokerConfiguration.getName(), f);
                                    // 给其他broker广播
                                    SubscriptionMessage message = new SubscriptionMessage();
                                    message.setType(TYPE_UNSUBSCIRBE);
                                    message.setBroker(brokerConfiguration.getName());
                                    message.setQos(q);
                                    message.setCleanSession(true);
                                    message.setFilter(f);
                                    redisTemplate.convertAndSend(CHANNEL_SUBSCRIPTION, JSON.toJSONString(message));
                                }
                            });
                        }

                    }
                }
            } catch (Exception e) {
                logger.error("error occured when disconnect client {}, cause:{}", e.getMessage());
            }
        }
        // delete the client from zookeeper
        zkHelper.removeClient(mqttId);
    }

    private boolean canDeleteFromRedis1(String filter){
        LinkedList<SubscriptionTree.ClientInfo<Tree1Client>> clients = tree1.getClients(filter);
        if (clients.isEmpty()){
            return true;
        }
        SubscriptionTree.ClientInfo<Tree1Client> c = clients.poll();
        while (c != null){
            // 如果还存在以CleanSession = true的订阅，则还不能从redis1中删除
            if (c.getClient().isCleanSession()){
                return false;
            }
            c = clients.poll();
        }
        return true;
    }
}


